package com.yl.flink.processor;

import cn.hutool.core.util.StrUtil;
import com.yl.constant.Const;
import com.yl.entity.MultiDataEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBPoint;

import java.util.HashMap;
import java.util.Map;

/**
 * @author wlf
 * @since 2022/8/18
 */
@Slf4j
public class RawPointMapFunc implements MapFunction<MultiDataEntity, InfluxDBPoint> {

    /**
     * @param multiDataEntity 流元素
     * @return InfluxDBPoint influxdb一个数据点
     */
    @Override
    public InfluxDBPoint map(MultiDataEntity multiDataEntity) throws Exception {
        // 表名
        String tbl = StrUtil.format(Const.TPL_TBL_RAW, multiDataEntity.getTargetType().getTag().toLowerCase());
        // 数据时间
        long timestamp = multiDataEntity.getPickTime();
        // tags
        HashMap<String, String> tags = new HashMap<>();
        tags.put(Const.S_TID, multiDataEntity.getTarget().getId().toString());
        // fields
        HashMap<String, Object> fields = new HashMap<>();
        Map<Integer, Double> rawValues = multiDataEntity.getRawValues();
        for (Integer qc : rawValues.keySet()) {
            fields.put(StrUtil.format(Const.TPL_FIELD, qc), rawValues.get(qc));
        }
        return new InfluxDBPoint(tbl, timestamp, tags, fields);
    }
}
